use eastmoney::stock::stock_intraday_transaction::StockTransaction;
use eastmoney::stock::EastmoneyTick;
use fudata::db;
use fudata::model;
use fudata::model::kline::Klt;
use fudata::model::security::Security;
use fudata::model::todo::Todo;
use fudata::model::todo::TodoParameter;
use fudata::model::todo::TodoStatus;
use fudata::model::todo::TodoType;
use futures::TryStreamExt;
use log::error;
use log::trace;
use log::warn;
use surrealdb::sql::Thing;
use std::collections::HashMap;
use fudata::model::tick::Tick;
use std::sync::Arc;
use anyhow::Result;
use anyhow::anyhow;
use log::info;
use crossbeam_deque::Injector;
use eventsource_client::Event;
use tokio::sync::Mutex;
use tokio::sync::Semaphore;
use crate::todo::sse_minute1klines_todo;
use crate::todo::RECEIVE_TIMEOUT_MS;
use std::time::Duration;
use super::sse_tick3_todo;
use super::MAX_WORKERS;

lazy_static::lazy_static! {
    // tick3
    static ref CONNECTION_POOL_TICK3: Arc<Mutex<HashMap<String, SseClientWrapper>>> = Arc::new(Mutex::new(HashMap::new()));
    static ref SEMAPHORE_TICK3: Arc<Semaphore> = Arc::new(Semaphore::new(100)); // Adjust the number according to your resource limitations
    static ref SSE_WORKERS_TICK3 : Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()> >>> = Arc::new(Mutex::new(HashMap::new()));
// 1minute1klines
    static ref CONNECTION_POOL_M001KLINES: Arc<Mutex<HashMap<String, SseClientWrapper>>> = Arc::new(Mutex::new(HashMap::new()));
    static ref SEMAPHORE_M001KLINES: Arc<Semaphore> = Arc::new(Semaphore::new(100)); // Adjust the number according to your resource limitations
    static ref SSE_WORKERS_M001KLINES : Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()> >>> = Arc::new(Mutex::new(HashMap::new()));
}

pub struct TodoInjector(pub Injector<Todo>);

#[derive(Debug, Clone)]
pub struct SseEventWrapper {
    pub event: Event,
    pub security_id: String,
    pub type_: EastmoneySseType,
    pub event_time: Option<chrono::DateTime<chrono::Local>>,
}

// Create a worker pool using Tokio
pub fn start_todo_worker_pool(num_workers: u8) -> Arc<TodoInjector> {
    let sse_event_injector = start_event_worker_pool(MAX_WORKERS);

    let injector = Arc::new(TodoInjector(Injector::new()));

    // let session_clone = Arc::clone(&session);

    // Spawn workers directly using tokio::spawn
    for worker_no in 0..num_workers {
        let tdi_1 = Arc::clone(&injector);
        let sei_1 = Arc::clone(&sse_event_injector);
        tokio::spawn(async move {
            loop {
                let task = tdi_1.0.steal().success();
                let sei_2 = Arc::clone(&sei_1);
                match task {
                    Some(todo) => {
                        // Process the todo using the handle_query function
                        let _sample = handle_sse_todo(worker_no, &todo, sei_2).await;
                        // Do something with the sample...
                    }
                    None => {
                        trace!("(worker_sse) Workder[{worker_no}] steal no todo, wait {}ms", RECEIVE_TIMEOUT_MS);
                        tokio::time::sleep(
                            std::time::Duration::from_millis(RECEIVE_TIMEOUT_MS)
                        ).await;
                    }
                }
            }
        });
    }

    injector
}

pub struct SseEventInjector(pub Injector<SseEventWrapper>);
// Create a sse-event-worker pool using Tokio
pub fn start_event_worker_pool(num_workers: u8) -> Arc<SseEventInjector> {
    let sei = Arc::new(SseEventInjector(Injector::new()));

    // Spawn workers directly using tokio::spawn
    for worker_no in 0..num_workers {
        let sei_1 = Arc::clone(&sei);
        tokio::spawn(async move {
            loop {
                let task = sei_1.0.steal().success();

                match task {
                    Some(sew) => {
                        //SseEventWrapper
                        // Process the todo using the handle_query function
                        let _sample = handle_sse_event(worker_no, &sew).await;
                        // Do something with the sample...
                    }
                    None => {
                        trace!("(start_event_worker_pool) Workder[{worker_no}] steal no todo, wait {}ms", RECEIVE_TIMEOUT_MS);
                        tokio::time::sleep(
                            std::time::Duration::from_millis(RECEIVE_TIMEOUT_MS)
                        ).await;
                    }
                }
            }
        });
    }

    sei
}

async fn handle_sse_event(worker_no: u8, sew: &SseEventWrapper) -> Result<()> {
    let event = &sew.event;
    let security_id = &sew.security_id;
    let event_data = sew.event.data.clone();
    let limit = if event_data.len() > 510 { 500 } else { event_data.len() };
    info!(
        "(handle_sse_event) Worker[{worker_no}] {}.sse_event:  {:?}",
        sew.security_id,
        &event_data[..limit]
    );

    match sew.type_ {
        EastmoneySseType::TICK3 => {
            let dd: StockTransaction = serde_json::from_str(&event_data).unwrap_or_default();
            if let Some(eastmoney_ticks) = dd.data.ticks() {
                let ticks: Vec<Tick> = convert_to_ticks(security_id, &eastmoney_ticks);
                let _ = db::tick::insert_all(&ticks).await;
            } else {
                println!("<<< (handle_sse_event)ticks [{}]: None", security_id);
            }
        }

        EastmoneySseType::MINUTE1 => {
            warn!(
                r#"TODO: handle_sse_todo: EastmoneySseType::MINUTE1, \n 
                This feature is not good enough as duplicated SSEs from server, \n 
                just use the ticks to derieve the 1minute klines."#
            );
        }
    }

    Ok(())
}

fn convert_to_ticks(security_id: &str, eastmoney_ticks: &Vec<EastmoneyTick>) -> Vec<Tick> {
    let mut ticks = vec![];

    for tick in eastmoney_ticks {
        let tick = Tick {
            id: Some(Thing::from(("tick", Tick::id_from(security_id, &tick.time.into()).as_str()))),
            time: tick.time.into(),
            price: tick.price,
            volume: tick.volume,
            trade_count: tick.trade_count,
            s1b2j4: tick.s1b2j4,
            ..Default::default()
        };

        ticks.push(tick);
    }

    ticks
}

// Define the handle_sse_todo function
async fn handle_sse_todo(
    worker_no: u8,
    todo: &Todo,
    sse_event_injector: Arc<SseEventInjector>
) -> Result<()> {
    if let Some(type_) = &todo.type_ {
        match type_ {
            TodoType::SseTick3 => {
                info!("(handle_sse_todo) Worker[{worker_no}] sse_tick3:  {:?}", todo.id);
                let _ = sse_tick3_todo::sse_tick3(
                    worker_no,
                    todo,
                    sse_event_injector,
                    SSE_WORKERS_TICK3.clone(),
                    SEMAPHORE_TICK3.clone(),
                    CONNECTION_POOL_TICK3.clone()
                ).await;
            }

            TodoType::SseM001Klines => {
                info!("(handle_sse_todo) Worker[{worker_no}] sse_minute1klines:  {:?}", todo.id);
                let _ = sse_minute1klines_todo::sse_minute1klines(
                    worker_no,
                    todo,
                    sse_event_injector,
                    SSE_WORKERS_M001KLINES.clone(),
                    SEMAPHORE_M001KLINES.clone(),
                    CONNECTION_POOL_M001KLINES.clone()
                ).await;
            }

            _ => {
                error!("(handle_sse_todo) unknown todo.type_:{}", type_);
            }
        }
    }

    Ok(())
}

pub async fn generate_sse_todos() {
    if false {
        // download kzz
        let _ = sse_tick3_todo::generate_todo_sse_kzz_tick3().await;
        // let _ = sse_minute1klines_todo::generate_todo_sse_kzz_minute1klines().await;
        // download etf
        // let _ = crawl_etf_todo::generate_todo_crawl_etf_list().await;
        // let _ = crawl_etf_todo::generate_todos_crawl_etf_klines().await;
        // download stock
        // let _ = crawl_stock_todo::generate_todo_crawl_stock_list().await;
        // let _ = crawl_stock_todo::generate_todos_crawl_stock_klines().await;
    } else {
        // download kzz
        let _ = sse_tick3_todo::generate_todo_sse_kzz_tick3().await;
        let _ = sse_minute1klines_todo::generate_todo_sse_kzz_minute1klines().await;
    }
}

pub async fn get_oldest_sse_todos() -> Result<Vec<Todo>> {
    // let _type_ = Some(vec![TodoType::SseM001Klines]); //None;
    let type_ = Some(vec![TodoType::SseTick3]); //None;
    let status = Some(TodoStatus::Pending);
    // Select todos and push them onto the queue if not already enqueued
    db::todo::select(&type_, &status, true, 0, MAX_WORKERS as u16).await
}

pub async fn refresh_kline_with_ticks(){
    let rs = db::kline::refresh_kline_with_ticks("SH.KZZ.110052",&Klt::M005).await;
}

#[derive(Debug, Clone)]
pub enum EastmoneySseType {
    /// Level1 每三秒一次的成交推送
    TICK3,
    /// 1分钟K线推送
    MINUTE1,
}

/// We define a CONNECTION_POOL as a shared resource protected by a mutex to store connections for each security ID.
/// The get_or_create_connection function checks if a connection exists for the security ID and creates one if it does not.
/// The distribute_evenly function distributes securities evenly among hosts.
/// The Semaphore limits the number of concurrent connections to prevent overloading the system.
/// The sse_worker function handles SSE events for a given security, acquiring a permit from the semaphore before starting to ensure resource limits are respected.

#[derive(Clone)]
pub struct SseClientWrapper {
    security_id: String,
    eastmoney_security_id: String,
    host: String,
    type_: EastmoneySseType,
    client: Arc<dyn eventsource_client::Client + Send + Sync + 'static>,
}

impl SseClientWrapper {
    pub fn new(
        security_id: &str,
        eastmoney_security_id: &str,
        host: &str,
        type_: EastmoneySseType,
        client: Arc<dyn eventsource_client::Client + Send + Sync + 'static>
    ) -> Self {
        SseClientWrapper {
            security_id: security_id.to_string(),
            eastmoney_security_id: eastmoney_security_id.to_string(),
            host: host.to_string(),
            type_,
            client,
        }
    }

    fn client(&self) -> &dyn eventsource_client::Client {
        &*self.client
    }
}

/// level 1 每3秒一次成交SSE推送
pub async fn sse_worker(
    worker_no: u16,
    client_wrapper: &SseClientWrapper,
    sse_event_injector: Arc<SseEventInjector>,
    semaphore: Arc<Semaphore>
) -> Result<()> {
    let _permit = semaphore.acquire().await.unwrap();
    let secid = &client_wrapper.eastmoney_security_id;

    let eventsource_client = client_wrapper.client();
    let mut eventsource_stream = eventsource_client
        .stream()
        .map_ok(|event| {
            match event {
                eventsource_client::SSE::Event(ev) => {
                    // println!("\n[{secid}] got an event: {} {}", ev.event_type, ev.data);
                    Some(ev)
                }
                eventsource_client::SSE::Comment(_comment) => {
                    println!("\n[{secid}] got a comment: \n{}", _comment);
                    None
                }
            }
        })
        .map_err(|err| eprintln!("\n (sse_worker) [{secid}] error streaming events: {:?}", err));

    // loop the stream
    loop {
        let rs_next = eventsource_stream.try_next().await;
        match rs_next {
            Ok(ooe) => {
                if let Some(Some(event)) = ooe {
                    let event_data = event.data.clone();
                    let limit = if event_data.len() > 510 { 500 } else { event_data.len() };
                    info!(
                        "(sse_worker) Worker[{worker_no}] sse_event_injector.0.push {}.sse_event:  {:?}",
                        client_wrapper.eastmoney_security_id,
                        &event_data[..limit]
                    );
                    sse_event_injector.0.push(SseEventWrapper {
                        event,
                        security_id: client_wrapper.security_id.clone(),
                        type_: client_wrapper.type_.clone(),
                        event_time: None,
                    });
                }
            }
            Err(e) => {
                error!("eventsource_stream.try_next().await {:?}", e);
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }
}

pub async fn get_todo_sse_parameters(security: &Security) -> Result<Vec<TodoParameter>> {
    let security_id = &security.id_raw();
    let mut eastmoney_security_id = format!("1.{}", &security.symbol);
    if security_id.starts_with("SH") {
        eastmoney_security_id = format!("1.{}", &security.symbol);
    } else if security_id.starts_with("SZ") {
        eastmoney_security_id = format!("0.{}", &security.symbol);
    } else {
        Err(
            anyhow!("(get_todo_sse_parameters) Unknown eastmoney_security_id from security_id:{}", security_id)
        )?;
    }

    return Ok(
        [
            TodoParameter {
                name: "security_id".to_string(),
                value: security_id.to_string(),
            },
            TodoParameter {
                name: "eastmoney_security_id".to_string(),
                value: eastmoney_security_id.to_string(),
            },
        ].to_vec()
    );
}

#[cfg(test)]
mod test {
    use crate::todo::worker_sse;

    #[tokio::test]
    async fn test_get_oldest_sse_todos() {
        pretty_env_logger::init_timed();
        let _ = fudata::db::init().await.unwrap();

        let rs = worker_sse::get_oldest_sse_todos().await;

        println!("rs: {:?}", rs);
    }
}
